package com.kvn.idpt;

import com.alibaba.fastjson.JSON;
import com.caucho.hessian.io.Hessian2Input;
import com.caucho.hessian.io.Hessian2Output;
import com.caucho.hessian.io.SerializerFactory;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.context.expression.MethodBasedEvaluationContext;
import org.springframework.core.DefaultParameterNameDiscoverer;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.jdbc.datasource.ConnectionHolder;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.sql.DataSource;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.reflect.Method;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * Created by wangzhiyuan on 2019/1/10
 */
@Slf4j
@Aspect
public class IdptTransactionAspect {
    public static final String IDPT_DDL = "CREATE TABLE if not EXISTS `_idempotent` (\n" +
            "  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键',\n" +
            "  `biz` varchar(64) NOT NULL COMMENT '业务标识',\n" +
            "  `idempotent_key` varchar(64) NOT NULL COMMENT '幂等key',\n" +
            "  `return_info` text COMMENT '返回信息',\n" +
            "  `return_byte` blob COMMENT '返回信息的二进制存储',\n" +
            "  `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',\n" +
            "  PRIMARY KEY (`id`),\n" +
            "  UNIQUE KEY `uniq_biz_idpt` (`biz`,`idempotent_key`)\n" +
            ") ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8";
    public static final String IDPT_QUERY = "SELECT return_info, return_byte FROM _idempotent WHERE biz= ? AND idempotent_key = ?";
    public static final String IDPT_SAVE = "INSERT INTO _idempotent (biz, idempotent_key, return_info, return_byte) VALUES (?, ?, ?, ?)";

    @Resource
    private DataSource dataSource;
    private boolean ddlAuto;

    public void setDdlAuto(boolean ddlAuto) {
        this.ddlAuto = ddlAuto;
    }

    @PostConstruct
    public void init(){
        if (ddlAuto) {
            try (Connection con = dataSource.getConnection(); PreparedStatement ps = con.prepareStatement(IDPT_DDL)) {
                ps.execute();
            } catch (SQLException e) {
                log.error("自动生成表失败", e);
            }
        }
    }

    @Pointcut("@annotation(com.kvn.idpt.IdptTransactional) && @annotation(org.springframework.transaction.annotation.Transactional)")
    public void idptPointcut(){}

    @Around("idptPointcut()")
    public Object around(ProceedingJoinPoint pjp) throws Throwable {
        // 获取幂等信息
        IdempotentInfo idempotentInfo = getIdempodentInfo(pjp);
        log.info("[keyExp, biz, idempotentKey] is [{}, {}, {}]", idempotentInfo.keyExp, idempotentInfo.biz, idempotentInfo.idempotentKey);

        ConnectionHolder connectionHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(dataSource);
        Connection connection = connectionHolder.getConnection();
        //
        ResultHolder resultHolder = getLastReturn(idempotentInfo, connection);
        if (resultHolder != null) {
            log.info("hit idempotent, return idempotent cache");
            return resultHolder.getResult();
        }

        Object result = pjp.proceed();
        serializeResult(result, idempotentInfo, connection);
        return result;
    }

    /**
     * 存储幂等结果
     * @param rlt
     * @param idempotentInfo
     * @param connection
     */
    private void serializeResult(Object rlt, IdempotentInfo idempotentInfo, Connection connection) {
        try (PreparedStatement ps = connection.prepareStatement(IDPT_SAVE)) {
            ps.setString(1, idempotentInfo.biz);
            ps.setString(2, idempotentInfo.idempotentKey);
            // 返回值为null || void方法
            if(rlt == null){
                ps.setString(3, null);
                ps.setBytes(4, null);
            } else {
                ByteArrayOutputStream os = new ByteArrayOutputStream();
                Hessian2Output h2o = new Hessian2Output(os);
                try {
                    h2o.writeObject(rlt);
                    h2o.flushBuffer();
                } catch (IOException e) {
                    throw new RuntimeException("幂等结果序列化失败", e);
                }
                ps.setString(3, JSON.toJSONString(rlt));
                ps.setBytes(4, os.toByteArray());
            }

            int count = ps.executeUpdate();
            log.info("serializeResult-save idempotent count:{}", count);
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    private ResultHolder getLastReturn(IdempotentInfo idempotentInfo, Connection con) {
        ResultSet rs = null;
        try (PreparedStatement ps = con.prepareStatement(IDPT_QUERY)) {
            ps.setString(1, idempotentInfo.biz);
            ps.setString(2, idempotentInfo.idempotentKey);
            rs = ps.executeQuery();
            if (rs.next()) {
                String return_info = (String) rs.getObject(1);
                byte[] return_byte = (byte[]) rs.getObject(2);
                return new ResultHolder(idempotentInfo.setReturnInfo(return_info).setReturnByte(return_byte).deserialize());
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        } finally {
            try {
                if (!rs.isClosed()) {
                    rs.close();
                }
            } catch (SQLException e) {
                log.error("ResultSet close error", e);
            }
        }
        // 首次调用，没有幂等记录
        return null;
    }

    private IdempotentInfo getIdempodentInfo(ProceedingJoinPoint pjp) {
        MethodSignature signature = (MethodSignature) pjp.getSignature();
        Method method = signature.getMethod();
        IdptTransactional idpt = method.getAnnotation(IdptTransactional.class);
        String keyExp = idpt.key();
        String idempotentKey = keyExp;
        if (keyExp == null || keyExp.length() <= 0) {
            throw new RuntimeException("指定的幂等 key 为空");
        }
        // 使用 spEL
        if (keyExp.contains("#")) {
            ExpressionParser parser = new SpelExpressionParser();
            StandardEvaluationContext context = new MethodBasedEvaluationContext(pjp.getTarget(), method, pjp.getArgs(), new DefaultParameterNameDiscoverer());
            idempotentKey = parser.parseExpression(keyExp).getValue(context, String.class);
        }

        return new IdempotentInfo().setBiz(idpt.biz()).setKeyExp(keyExp).setIdempotentKey(idempotentKey).setReturnClass(method.getReturnType());
    }

    @Data
    @Accessors(chain = true)
    private static class IdempotentInfo {
        /**
         * 业务标识
         */
        private String biz;

        /**
         * 防重 key
         */
        private String idempotentKey;

        /**
         * 防重 key 表达式（spEL）
         */
        private String keyExp;
        /**
         * 返回类型
         */
        private Class<?> returnClass;

        /**
         * 返回信息
         */
        private String returnInfo;

        /**
         * 返回信息的二进制存储
         */
        private byte[] returnByte;

        public Object deserialize(){
            if (returnClass == Void.TYPE) { // void 方法
                return null;
            }
            ByteArrayInputStream is = new ByteArrayInputStream(returnByte);
            Hessian2Input h2i = new Hessian2Input(is);
            h2i.setSerializerFactory(new SerializerFactory());
            try {
                return h2i.readObject(returnClass);
            } catch (Exception e) {
                throw new RuntimeException("反序列化失败");
            } finally {
                try {
                    is.close();
                } catch (Exception e) {
                    log.error("ByteArrayInputStream close error!", e);
                }
            }
        }
    }

    /**
     * 需要幂等方法返回值可能为：1. void 2. 非 void，但是为 null 3. 非 void，非 null
     */
    private static class ResultHolder {
        private Object result;

        public ResultHolder(Object result) {
            this.result = result;
        }

        public Object getResult() {
            return result;
        }
    }

}
